﻿using RdKafka;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace test.kafka
{
    public class KafkaConsumer
    {
        private readonly ConsumerConfig _consumerConfig;
        public KafkaConsumer(ConsumerConfig consumerConfig)
        {
            _consumerConfig = consumerConfig;
        }

        public void Consume(Func<string,string> handler) 
        {
            //配置消费者组
            var config = new Config() { GroupId = _consumerConfig.GroupId };
            using (var consumer = new EventConsumer(config, _consumerConfig.KafkaServer))
            {
                //注册一个事件
                consumer.OnMessage += (obj, msg) =>
                {
                    string message = Encoding.UTF8.GetString(msg.Payload, 0, msg.Payload.Length);
                    Console.WriteLine($"Topic: {msg.Topic} Partition: {msg.Partition} Offset: {msg.Offset} {message}");
                    handler(message);
                };

                //订阅一个或者多个Topic
                consumer.Subscribe(_consumerConfig.Topics);

                //启动
                consumer.Start();
                Console.WriteLine("Started consumer, press enter to stop consuming");
            }
        }
    }
}
